package com.yong.activemq.topic;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;
import java.io.IOException;

public class JmsConsumerTopic {
    static final String DEFAULT_URL = "tcp://localhost:61616";
    static final String TOPIC_NAME = "topic-01";

    /**
     * 发布/订阅消息传递域的特点如下：
     * （1）生产者将消息发布到topic中，每个消息可以有多个消费者，属于1：N的关系
     * （2）生产者和消费者之间有时间上的相关性。订阅某个主题的消费者只能消费自它订阅之后发布的消息
     * （3）生产者生产时，topic不保存消息它是无状态的不落地，假如无人订阅就去生产，那就是一条废消息，
     * 所以，一般先启动消费者再启动生产者。
     * <p>
     * JMS规范允许客户创建持久订阅，这在一定程度上放松了时间上的相关性要求。
     * 持久订阅允许消费者消费它在未处于激活状态时发送的消息。
     * 一句话，好比我们的微信公众号订阅
     *
     * @param args
     * @throws JMSException
     */
    public static void main(String[] args) throws JMSException, IOException {

        System.out.println("我是2号消费者");

        // 1、创建连接工厂
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(DEFAULT_URL);

        // 2、通过连接工厂创建connection，并且启动连接
        Connection connection = activeMQConnectionFactory.createConnection();
        connection.start();

        // 3、通过connection创建session
        // 第一个参数：事务； 第二个参数：签收
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // 4、通过session创建destination；topic
        Topic topic = session.createTopic(TOPIC_NAME);

        // 5、通过session创建consumer消费者
        MessageConsumer messageConsumer = session.createConsumer(topic);

        /*
        第一种访问方式：
        同步阻塞方式（receive()）
        订阅者或接收者调用messageConsumer的receive()方法来接收消息，receive方法在能够接收到消息之前(或超时之前)将一直阻塞.

        //记住：发送的是什么格式，自然接收的格式也要跟发送的格式一样
        while(true) {
            //不带时间的receive()方法，说明这方法一直在等待新消息。
            //TextMessage receive = (TextMessage) messageConsumer.receive();
            //带时间的receive()方法，说明该方法一旦到了时间就立刻结束接收。
            TextMessage receive = (TextMessage) messageConsumer.receive(5000L);

            if (receive != null){
                System.out.println("消费到内容："+receive.getText());
            }else{
                break;
            }

        }

        // 关闭所有连接
        messageConsumer.close();
        session.close();
        connection.close();*/


        /**
         *   第二种访问方式：
         *      能够解决消费者的监听，通过监听的方式消费消息
         */
        messageConsumer.setMessageListener(message -> {
            if (null != message && message instanceof TextMessage) {
                TextMessage textMessage = (TextMessage) message;
                try {
                    System.out.println("****消费者接收到消息:" + textMessage.getText());
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });

        System.in.read();  //保证控制台不关
        messageConsumer.close();
        session.close();
        connection.close();
    }
}
